package com.sunzm.spark.core

import com.alibaba.fastjson.JSON
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object RDDJOINDemo {
  def main(args: Array[String]): Unit = {
    //准备执行环境
    val conf = new SparkConf().setAppName("RDD示例")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)

    //读取订单数据
    val orderRDD: RDD[String] = sc.textFile("data/spark/rdd/order.txt")
    val userRDD: RDD[String] = sc.textFile("data/spark/rdd/user.txt")


    //变成key-value形式的RDD
    val orderMapRDD: RDD[(String, String)] = orderRDD.map(line => {
      val jSONObject = JSON.parseObject(line)
      val userId = jSONObject.getString("userId")
      //K 是userId, v line
      (userId, line)
    })


    val userMapRDD: RDD[(String, String)] = userRDD.map(line => {
      val jSONObject = JSON.parseObject(line)
      val userId = jSONObject.getString("userId")
      val userName = jSONObject.getString("userName")

      (userId, userName)
    })

    val joinRDD: RDD[(String, (String, String))] = orderMapRDD.join(userMapRDD)

    val resultRDD: RDD[String] = joinRDD.map(t => {
      //val userId = t._1
      val oderInfo: String = t._2._1
      val userName: String = t._2._2

      val jSONObject = JSON.parseObject(oderInfo)

      jSONObject.put("userName", userName);

      jSONObject.toJSONString
    })

    resultRDD.foreach(line => {
      println(line)
    })

    //关闭
    sc.stop()
  }
}
